package zk_task

import (
	"EasyCollect/global"
	"github.com/samuel/go-zookeeper/zk"
	"time"
	"fmt"
	"EasyCollect/monitor"
	"EasyCollect/util"
)

const (
	rootNode= "/easycollect" //根节点
	goroutine="goroutine"//协程数
	memory="memory"//内存
	taskId="taskId"//任务id
	taskName="taskName"//任务名称
	taskCommand="taskCommond"//任务注释
	cron="cron"//定时任务表达式
	thread="thread"//任务启用线程数
	isFlow="isFlow"//是否是流式执行
	speed="speed" //流式任务上下次间隔时间
	shardingCount="shardingCount"//分片数
	shardingColumn="ShardingColumn"//分片键
	srcIp="srcIp" //注释略，以下内容注释，见config
	srcPort="srcPort"
	srcCharacterEncoding="srcCharacterEncoding"
	srcDatabase="srcDatabase"
	srcTable="srcTable"
	srcLimit="srcLimit"
	srcKey="srcKey"
	srcCondition="srcCondition"
	srcParameter="srcParameter"
	srcQuery="srcQuery"
	srcDelete="srcDelete"
	targetIp="targetIp"
	targetPort="targetPort"
	targetCharacterEncoding="targetCharacterEncoding"
	targetDatabase="targetDatabase"
	targetTable="targetTable"
	targetType="targetType"
	targetColumnMapping="targetColumnMapping"
	targetUpdateCondition="targetUpdateCondition"
	excludeColumn="excludeColumn"
	open="open"//一个开关，用户从zk上手动触发修改为任意值后，手动触发定时任务
	stop="stop"//用于对正在执行的定时任务进行暂停和启动

)

/**
	开始配置每一个任务的具体配置项目，并注册监听
 */
func createChildSessionNodeIfExists(task *global.Task) {

    go createIfNotExists(task)

	//针对某些热配置有意义的属性，进行了监听和回调
	/**
		手动触发定时任务
	 */
	WatcherOpen :=&taskWatcherAndNotify{make(chan (<-chan zk.Event)),task,getTaskConfigNode(task.TaskName,open)}
	WatcherOpen.StartWatchAndNotify()
	/**
		暂停定时任务
	 */
	WatcherStop :=&taskWatcherAndNotify{make(chan (<-chan zk.Event)),task,getTaskConfigNode(task.TaskName,stop)}
	WatcherStop.StartWatchAndNotify()

}
/**
	如果配置节点不存在，就创建
	采用心跳检测的方式,原因：程序重启期间，zk会删除临时节点，因此，只创建一次节点可能会失效
 */
func createIfNotExists(task *global.Task) {
	for {

		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,taskId),util.ToString(task.TaskId))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,taskName),util.ToString(task.TaskName))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,taskCommand),util.ToString(task.TaskCommand))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,cron),util.ToString(task.Cron))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,thread),util.ToString(task.Thread))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,isFlow),util.ToString(task.IsFlow))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,speed),util.ToString(task.Speed))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,shardingCount),util.ToString(task.ShardingCount))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,shardingColumn),util.ToString(task.ShardingColumn))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,srcIp),util.ToString(task.SrcIp))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,srcPort),util.ToString(task.SrcPort))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,srcCharacterEncoding),util.ToString(task.SrcCharacterEncoding))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,srcDatabase),util.ToString(task.SrcDatabase))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,srcTable),util.ToString(task.SrcTable))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,srcLimit),util.ToString(task.SrcLimit))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,srcKey),util.ToString(task.SrcKey))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,srcCondition),util.ToString(task.SrcCondition))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,srcParameter),util.ToString(task.SrcParameter))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,srcQuery),util.ToString(task.SrcQuery))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,srcDelete),util.ToString(task.SrcDelete))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,targetIp),util.ToString(task.TargetIp))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,targetPort),util.ToString(task.TargetPort))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,targetCharacterEncoding),util.ToString(task.TargetCharacterEncoding))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,targetDatabase),util.ToString(task.TargetDatabase))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,targetTable),util.ToString(task.TargetTable))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,targetType),util.ToString(task.TargetType))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,targetColumnMapping),util.ToString(task.TargetColumnMapping))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,targetUpdateCondition),util.ToString(task.TargetUpdateCondition))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,excludeColumn),util.ToString(task.ExcludeColumn))

		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,open),"123456")
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,stop),"false")

		time.Sleep(30*time.Second)
	}
}
/**
	在应用的namespace下创建协程数量和内存利用率的信息
 */
func createIfNoExistsMemAndGetGoroutine() {
	for{
		createSetSessionNode(getTaskNode(goroutine),fmt.Sprintf("%v",monitor.GetGoroutine()))
		v1,v2,v3,v4:= monitor.GetMemoryMb()
		createSetSessionNode(getTaskNode(memory),fmt.Sprintf("%s,%s,%s,%s",v1,v2,v3,v4))

		time.Sleep(10*time.Second)
	}
}
